package twitterhadoop.app;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import twitterhadoop.hadoop.io.TextArrayWritable;


public class Reduce extends Reducer<ClosedItemSet, TextArrayWritable, Text, Text> {
	List<ClosedItemSet> closedItemsets = new ArrayList<ClosedItemSet>();
	
	public void reduce(ClosedItemSet key, Iterable<TextArrayWritable> values, Context context) throws IOException, InterruptedException {
		
			/* Accumulate the transactions for this key */
			List<String[]> transactions = new ArrayList<String[]>();		
		
			/* Convert from Writable String (Text) to plain String */
			for (TextArrayWritable nextTransaction : values) {
				String[] textToString = nextTransaction.toArray();
				// YGYG no need: 
				// context.write(new Text(key.toString()), new Text(nextTransaction.toString()));
				transactions.add(textToString);
			}
		
			/* Get the minimum support from the context */
			int minSupport = context.getConfiguration().getInt(TwitterClosedItemSet.MIN_SUPP_PARAMETER_NAME, Integer.MAX_VALUE);
			
			/* Dismiss keys with less than minimum support */
			int support = transactions.size();
			if (support < minSupport) {
				return;
			}
		
			/* Intersect the transactions to find the closure on the key */
			String[] closure = intersection(transactions);
			
			/* Check if the key for this reducer is lexicographically smaller then the intersection.
			 * if so then this reducer output is valid.
			 * else ignore this reducer result to prevent duplicates. 
			 */
			//if (compareArrays(key.getGenerator(), closure) > 0) {
			//	return;
			//}
			String firstItem = addedItemInClosure(key.getItemset(), closure);
			if (firstItem != null) {
				if (key.getAddedItem().compareTo(firstItem) > 0) {
					return;
				}				
			}
			
			
			this.closedItemsets.add(new ClosedItemSet(key.getGenerator(), closure, support));			
			
			context.getCounter(InternalCounters.REDUCER_OUTPUT).increment(1);
			//context.write(out, null);
			
			// YGYG According to Dror'd observation, we emit the key
			// with the seed, which is used in the Mapper for duplicate elimination.
	}
	
	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
		Configuration conf = context.getConfiguration();
		int jobCounter = conf.getInt(TwitterClosedItemSet.JOB_COUNTER, 0);
		
		String pathString = String.format("output/closed.itemsets.%s/%s", jobCounter, context.getTaskAttemptID().getTaskID().getId());

		// Path outPath = new Path(conf.get(TwitterClosedItemSet.CLOSED_ITEMSETS_PATH_ATTR, "output/closed.itemsets." + jobCounter));
		Path outPath = new Path(pathString);
		FileSystem fs = FileSystem.get(conf);
		fs.delete(outPath, true);

		//final SequenceFile.Writer out = SequenceFile.createWriter(fs, context.getConfiguration(), outPath, ClusterCenter.class, IntWritable.class);
		FSDataOutputStream out = fs.create(outPath);
		// FSDataOutputStream out = fs.append(outPath);
		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));		

		/*
		Path outPath = new Path(conf.get(TwitterClosedItemSet.CLOSED_ITEMSETS_PATH_ATTR, "output/closed.itemsets" + "." + jobCounter));
		FileSystem fs = FileSystem.get(conf);
		fs.delete(outPath, true);
		
		// final SequenceFile.Writer out = SequenceFile.createWriter(fs, context.getConfiguration(), outPath, ClusterCenter.class, IntWritable.class);
		FSDataOutputStream out = fs.create(outPath);
		// FSDataOutputStream out = fs.append(outPath);
		BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));		
		*/
		
		for (ClosedItemSet itemset : this.closedItemsets) {
			writer.write(itemset.serialize());
			writer.write('\n');
		}
		writer.close();
		out.close();
	}	
	
	public static String[] intersection (List<String[]> vector) {
		if(vector.size() == 1) {
			return vector.get(0);
		}
		
		String[] tempRes = vector.get(0);
		for (int i = 1; i < vector.size() ; i++) {
			tempRes = intersection(tempRes, vector.get(i)); 
		}
		return tempRes;
	}

	@SuppressWarnings("unchecked")
	public static String[] intersection(String[] one, String[] two) {
		List<String> list_one = Arrays.asList(one);
	    List<String> list_two = Arrays.asList(two);

	    List<String> intersection = (List<String>) CollectionUtils.intersection(list_one, list_two);
	    Collections.sort(intersection);

	    return (String[]) intersection.toArray(new String[intersection.size()]);		
	}
	
	public static int compareArrays(String[] arr1, String[] arr2) {
		int comp = 0;
		int pos = 0;
		try {
			while (pos < arr1.length & pos < arr2.length) {
				comp = arr1[pos].compareTo(arr2[pos]);
				if (comp != 0 ) {
					return comp;
				}
				pos++;
			}
			
			return arr1.length - arr2.length;

		} catch (ArrayIndexOutOfBoundsException e) {
			e.printStackTrace();
		}
		return comp;
	}
	
	/**
	 * Returns the smallest item in the closure not appearing in the itemset. <br>
	 * For example, for input closure: {a,c,d,f} and itemset {a,c} returns d.
	 * 
	 * @param itemsetArray
	 * @param closureArray
	 * @return
	 */
	public static String addedItemInClosure(String[] itemsetArray, String[] closureArray) {
		String res;
		Set<String> itemset = new HashSet<String>(Arrays.asList(itemsetArray));
		Set<String> closure = new HashSet<String>(Arrays.asList(closureArray));
		
		closure.removeAll(itemset);
		
		List<String> diff = new ArrayList<String>(closure);
		
		if (diff.size() == 0) {
			res = null;
		} else {
			Collections.sort(diff);
			res = diff.get(0);
			
		}
		
		return res;
	}

}
	